package com.basker.pisces.orm.session;

import java.lang.reflect.Array;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.stream.Stream;

import org.apache.commons.lang3.ArrayUtils;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.dao.OptimisticLockingFailureException;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import com.basker.pisces.cache.ICache;
import com.basker.pisces.core.IDataContract;
import com.basker.pisces.core.RowContract;
import com.basker.pisces.core.datasource.DbInfo;
import com.basker.pisces.core.datasource.IDataSourceManager;
import com.basker.pisces.core.enums.DataObjectState;
import com.basker.pisces.core.exception.StaleObjectException;
import com.basker.pisces.core.iterate.DataCollectionIterateEvent;
import com.basker.pisces.core.iterate.DataObjectIterateEvent;
import com.basker.pisces.core.iterate.DataObjectIterator;
import com.basker.pisces.core.iterate.IDataObjectIterateHandler;
import com.basker.pisces.core.meta.ICollectionField;
import com.basker.pisces.core.meta.IDataField;
import com.basker.pisces.core.meta.IDataObjectField;
import com.basker.pisces.core.meta.IDataObjectMeta;
import com.basker.pisces.core.meta.IField;
import com.basker.pisces.core.meta.row.RowContractMeta;
import com.basker.pisces.core.meta.row.RowDataField;
import com.basker.pisces.core.meta.row.RowReferenceField;
import com.basker.pisces.core.utils.ConvertUtils;
import com.basker.pisces.orm.cache.DataContractCacheKeyHelper;
import com.basker.pisces.orm.cache.DataContractCacheManager;
import com.basker.pisces.orm.cache.DataContractCacheOption;
import com.basker.pisces.orm.cache.DataContractLocalCacheManager;
import com.basker.pisces.orm.cache.DataContractRedisCacheManager;
import com.basker.pisces.orm.cache.NullObject;
import com.basker.pisces.orm.data.DataSet;
import com.basker.pisces.orm.data.meta.DataSetMeta;
import com.basker.pisces.orm.data.meta.FieldMeta;
import com.basker.pisces.orm.data.meta.TableMeta;
import com.basker.pisces.orm.event.DataObjectWriteEvent;
import com.basker.pisces.orm.execute.ISqlExecutor;
import com.basker.pisces.orm.parser.ISqlParser;
import com.basker.pisces.resource.StringResources;
import com.basker.pisces.spring.SpringApplicationContext;
import com.basker.pisces.tx.TransactionHelper;
import com.basker.pisces.utils.PageUtils;

/**
 * 默认的数据库会话实例,内部封装的数据源是一个动态数据源对象。
 *
 * <p>
 * 如果当前实例需要被事务管控，那么保存、删除、查询、加载等方法内部会被事务控制。
 *
 * <p>
 * 由于使用的事spring事务管理器，如果外部已经开启事务，则会直接使用上下文的事务对象
 *
 * @author hangwen
 */
class DefaultSqlSession implements ISqlSession {

    private static final int[] EMPTY_ARGS_TYPE = new int[0];
    private static final Object[] EMPTY_ARGS = ArrayUtils.EMPTY_OBJECT_ARRAY;

    /**
     * 数据对象的缓存管理器
     *
     * @see DataContractLocalCacheManager
     * @see DataContractRedisCacheManager
     */
    private DataContractCacheManager cacheManager;

    /**
     * spring事件发布器
     */
    private ApplicationEventPublisher applicationEventPublisher;

    /**
     * 当前会员访问的数据源
     */
    private String dataSourceName;

    /**
     * sql执行器
     */
    private ISqlExecutor sqlExecutor;

    /**
     * 数据源管理器
     */
    private IDataSourceManager dataSourceManager;

    /**
     * spring事务模板
     */
    private TransactionTemplate transactionTemplate;

    /**
     * sql解析器
     */
    private ISqlParser sqlParser;

    /**
     * 数据保存后，该对象会迭代数据，重新设置其状态
     */
    private AcceptChangeHandler acceptChangeHandler = new AcceptChangeHandler();

    public DefaultSqlSession() {

    }

    @Override
    public void close() {
        this.dataSourceManager.removeContextDataSourceName();

        this.sqlExecutor = null;
        this.sqlParser = null;
        this.transactionTemplate = null;
    }

    @Override
    public void delete(IDataObjectMeta objectMeta, Object id) {
        Assert.notNull(id, "parameter 'id' is required");

        this.delete(objectMeta, new Object[]{id});
    }

    @Override
    public void delete(IDataObjectMeta objectMeta, Object[] ids) {
        Assert.notNull(objectMeta, "parameter 'objectMeta' is required");

        if (ArrayUtils.isEmpty(ids)) {
            return;
        }

        DataWriter dataWriter = new DataWriter(objectMeta, this.getSqlParser(), this.getSqlExecutor(),
                this.getTransactionTemplate(), _ids -> this.removeFromCache(objectMeta, _ids));
        dataWriter.delete(ids);

        // 发布数据写入事件
        DataObjectWriteEvent writeEvent = dataWriter.getWriteEvent();
        if (writeEvent != null) {
            publishWriteEvent(writeEvent);
        }
    }

    @Override
    public void execute(String sql) {
        Assert.hasText(sql, "parameter 'sql' is required");

        this.getSqlExecutor().execute(sql);
    }

    @Override
    public void fillReferenceFields(IDataObjectMeta objectMeta, Object[] dataObjects) {
        Assert.notNull(objectMeta, "parameter 'objectMeta' is required");

        if (ArrayUtils.isEmpty(dataObjects)) {
            return;
        }

        ReferenceFieldLoader loader = new ReferenceFieldLoader(this);
        loader.loadRefrenceFields(objectMeta, dataObjects);
    }

    public ApplicationEventPublisher getApplicationEventPublisher() {
        return applicationEventPublisher;
    }

    public DataContractCacheManager getCacheManager() {
        return cacheManager;
    }

    public IDataSourceManager getDataSourceManager() {
        return dataSourceManager;
    }

    public String getDataSourceName() {
        return dataSourceName;
    }

    @Override
    public ISqlParser getSqlParser() {
        return sqlParser;
    }

    @Override
    public Object load(IDataObjectMeta objectMeta, Object id, boolean useCache, boolean forUpdate) {
        Assert.notNull(objectMeta, "parameter 'objectMeta' is required");
        Assert.notNull(id, "parameter 'id' is required");

        Object[] objects = this.loadBatch(objectMeta, new Object[]{id}, useCache, forUpdate);
        if (ArrayUtils.isEmpty(objects)) {
            return null;
        }

        return objects[0];
    }

    @Override
    public Object[] loadBatch(IDataObjectMeta objectMeta, Object[] ids, boolean useCache, boolean forUpdate) {
        Assert.notNull(objectMeta, "parameter 'objectMeta' is required");

        if (ArrayUtils.isEmpty(ids)) {
            Class<?> dataObjectClass = objectMeta.getDataObjectClass();
            return (Object[]) Array.newInstance(dataObjectClass, 0);
        }

        if (useCache) {
            return this.loadBatchFromCache(objectMeta, ids, forUpdate);
        } else {
            return this.loadBatchFormDB(objectMeta, ids, forUpdate);
        }
    }

    @Override
    public List<Map<String, Object>> queryForList(String sql, Object[] args) {
        Assert.hasText(sql, "parameter 'sql' is required");

        return this.queryForList(sql, args, EMPTY_ARGS_TYPE);
    }

    @Override
    public List<Map<String, Object>> queryForList(String sql, Object[] args, int[] argTypes) {
        Assert.hasText(sql, "parameter 'sql' is required");

        return this.getSqlExecutor().queryForList(sql, args, argTypes);
    }

    @Override
    public Map<String, Object> queryForMap(String sql, Object[] args) {
        Assert.hasText(sql, "parameter 'sql' is required");

        return this.queryForMap(sql, args, EMPTY_ARGS_TYPE);
    }

    @Override
    public Map<String, Object> queryForMap(String sql, Object[] args, int[] argTypes) {
        Assert.hasText(sql, "parameter 'sql' is required");

        return this.getSqlExecutor().queryForMap(sql, args, argTypes);
    }

    @Override
    public <TResult> TResult queryForObject(Class<TResult> requiredType, String sql, Object[] args) {
        Assert.notNull(requiredType, "parameter 'requiredType' is required");
        Assert.hasText(sql, "parameter 'sql' is required");

        return this.queryForObject(requiredType, sql, args, EMPTY_ARGS_TYPE);
    }

    @Override
    public <TResult> TResult queryForObject(Class<TResult> requiredType, String sql, Object[] args, int[] argTypes) {
        Assert.notNull(requiredType, "parameter 'requiredType' is required");
        Assert.hasText(sql, "parameter 'sql' is required");

        return this.getSqlExecutor().queryForObject(requiredType, sql, args, argTypes);
    }

    @Override
    public Object[] queryForObjects(IDataObjectMeta objectMeta, String whereSql, Object[] args, boolean userCache,
                                    boolean forUpdate) {
        Assert.notNull(objectMeta, "parameter 'objectMeta' is required");

        return this.queryForObjects(objectMeta, whereSql, args, EMPTY_ARGS_TYPE, userCache, forUpdate);
    }

    @Override
    public Object[] queryForObjects(IDataObjectMeta objectMeta, String whereSql, Object[] args, int[] argTypes,
                                    boolean userCache, boolean forUpdate) {
        Assert.notNull(objectMeta, "parameter 'objectMeta' is required");

        if (StringUtils.isEmpty(whereSql) || args == null) {
            args = EMPTY_ARGS;
        }

        if (userCache) {
            return queryForObjectsFromCache(objectMeta, whereSql, forUpdate, args, argTypes);
        } else {
            return this.queryForObjectsFromDB(objectMeta, whereSql, forUpdate, args, argTypes);
        }
    }

    @Override
    public List<Map<String, Object>> queryForPageList(String sql, int pageIndex, int pageSize, Object[] args) {
        Assert.hasText(sql, "parameter 'sql' is required");
        Assert.isTrue(pageIndex >= 0, "parameter 'pageIndex' must >= 0");
        Assert.isTrue(pageSize > 0, "parameter 'pageSize' must > 0");

        String pageSql = this.getSqlParser().resolvePageSql(sql, pageIndex, pageSize);
        return this.getSqlExecutor().queryForList(pageSql, args, EMPTY_ARGS_TYPE);
    }

    @Override
    public List<Map<String, Object>> queryForPageList(String sql, int pageIndex, int pageSize, Object[] args,
                                                      int[] argTypes) {
        Assert.hasText(sql, "parameter 'sql' is required");
        Assert.isTrue(pageIndex >= 0, "parameter 'pageIndex' must >= 0");
        Assert.isTrue(pageSize > 0, "parameter 'pageSize' must > 0");

        String pageSql = this.getSqlParser().resolvePageSql(sql, pageIndex, pageSize);
        return this.getSqlExecutor().queryForList(pageSql, args, argTypes);
    }

    @Override
    public List<RowContract> queryForPageRows(RowContractMeta rowMeta, String sql, int pageIndex, int pageSize,
                                              Object[] args) {
        Assert.notNull(rowMeta, "parameter 'rowMeta' is required");
        Assert.hasText(sql, "parameter 'sql' is required");
        Assert.isTrue(pageIndex >= 0, "parameter 'pageIndex' must >= 0");
        Assert.isTrue(pageSize > 0, "parameter 'pageSize' must > 0");

        return this.innerQueryForRows(rowMeta, sql, pageIndex, pageSize, args, EMPTY_ARGS_TYPE);
    }

    @Override
    public List<RowContract> queryForPageRows(RowContractMeta rowMeta, String sql, int pageIndex, int pageSize,
                                              Object[] args, int[] argTypes) {
        Assert.notNull(rowMeta, "parameter 'rowMeta' is required");
        Assert.hasText(sql, "parameter 'sql' is required");
        Assert.isTrue(pageIndex >= 0, "parameter 'pageIndex' must >= 0");
        Assert.isTrue(pageSize > 0, "parameter 'pageSize' must > 0");

        return this.innerQueryForRows(rowMeta, sql, pageIndex, pageSize, args, argTypes);
    }

    @Override
    public List<RowContract> queryForRows(RowContractMeta rowMeta, String sql, Object[] args) {
        Assert.notNull(rowMeta, "parameter 'rowMeta' is required");
        Assert.hasText(sql, "parameter 'sql' is required");

        return this.innerQueryForRows(rowMeta, sql, -1, -1, args, EMPTY_ARGS_TYPE);
    }

    @Override
    public List<RowContract> queryForRows(RowContractMeta rowMeta, String sql, Object[] args, int[] argTypes) {
        Assert.notNull(rowMeta, "parameter 'rowMeta' is required");
        Assert.hasText(sql, "parameter 'sql' is required");

        return this.innerQueryForRows(rowMeta, sql, -1, -1, args, argTypes);
    }

    @Override
    public Object save(IDataObjectMeta objectMeta, Object dataObject) {
        Assert.notNull(objectMeta, "parameter 'objectMeta' is required");
        Assert.notNull(dataObject, "parameter 'dataObject' is required");

        Object[] ids = this.save(objectMeta, new Object[]{dataObject});
        return ids.length > 0 ? ids[0] : null;
    }

    @Override
    public Object[] save(IDataObjectMeta objectMeta, Object[] dataObjects) {
        Assert.notNull(objectMeta, "parameter 'objectMeta' is required");
        Assert.notEmpty(dataObjects, "parameter 'dataObjects' is required");

        for (Object dataObject : dataObjects) {
            Assert.notNull(dataObject, "parameter 'dataObjects' has null object");
        }

        // 保存对象的具体过程由DataWriter实例负责
        DataWriter dataWriter = new DataWriter(objectMeta, this.getSqlParser(), this.getSqlExecutor(),
                this.getTransactionTemplate(), ids -> this.removeFromCache(objectMeta, ids));

        Object[] ids = null;

        // 保存可能会抛出乐观锁失败异常
        try {
            ids = dataWriter.save(dataObjects);
        } catch (OptimisticLockingFailureException e) {
            String message = StringResources.get("pisces-orm.StaleObjectException", objectMeta.getDisplayName());
            throw new StaleObjectException(message, e);
        }

        // 更改数据的状态
        this.acceptChange(objectMeta, dataObjects);

        // 发布数据写入事件
        DataObjectWriteEvent writeEvent = dataWriter.getWriteEvent();
        if (writeEvent != null) {
            publishWriteEvent(writeEvent);
        }

        return ids;
    }

    public void setSqlParser(ISqlParser sqlParser) {
        this.sqlParser = sqlParser;
    }

    @Override
    public int update(String sql, Object[] args) {
        Assert.hasText(sql, "parameter 'sql' is required");

        return this.update(sql, args, EMPTY_ARGS_TYPE);
    }

    @Override
    public int update(String sql, Object[] args, int[] argTypes) {
        Assert.hasText(sql, "parameter 'sql' is required");

        return this.getSqlExecutor().update(sql, args, argTypes);
    }

    ISqlExecutor getSqlExecutor() {
        return sqlExecutor;
    }

    TransactionTemplate getTransactionTemplate() {
        return transactionTemplate;
    }

    void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
        this.applicationEventPublisher = applicationEventPublisher;
    }

    void setCacheManager(DataContractCacheManager cacheManager) {
        this.cacheManager = cacheManager;
    }

    void setDataSourceManager(IDataSourceManager dataSourceManager) {
        this.dataSourceManager = dataSourceManager;
    }

    void setDataSourceName(String dataSourceName) {
        this.dataSourceName = dataSourceName;

        // 设置上下文的数据源
        this.dataSourceManager.setContextDataSourceName(dataSourceName);
        // 设置对应数据源的sql解析器
        this.setSqlParser(dataSourceName, dataSourceManager);
    }

    void setSqlExecutor(ISqlExecutor sqlExecutor) {
        this.sqlExecutor = sqlExecutor;
    }

    void setTransactionTemplate(TransactionTemplate transactionTemplate) {
        this.transactionTemplate = transactionTemplate;
    }

    /**
     * 用acceptChangeHandler对象迭代所有dataObjects
     *
     * @param objectMeta
     * @param dataObjects
     * @see AcceptChangeHandler
     */
    private void acceptChange(IDataObjectMeta objectMeta, Object[] dataObjects) {
        DataObjectIterator iterator = new DataObjectIterator(dataObjects, objectMeta, acceptChangeHandler);

        if (TransactionHelper.inTransactionScope()) {
            TransactionHelper.doBeforeCommit((readOnly) -> {
                iterator.iterate();
            });
        } else {
            iterator.iterate();
        }
    }

    private void flatReferenceFields(RowContractMeta rowMeta, List<RowContract> rows) {
        Collection<IDataField> dataFields = rowMeta.getDataFields();

        for (IDataField dataField : dataFields) {
            RowDataField rowDataField = (RowDataField) dataField;
            String referenceFieldName = rowDataField.getHostTableFkFieldName();

            if (StringUtils.isEmpty(referenceFieldName)) {
                continue;
            }

            IField refField = rowMeta.getField(referenceFieldName);
            if (refField instanceof RowReferenceField) {
                IField referencedField = ((RowReferenceField) refField).getReferencedField(rowDataField.getName());

                for (RowContract row : rows) {
                    Object refObject = refField.getValue(row);

                    if (refObject != null) {
                        Object value = referencedField.getValue(refObject);

                        rowDataField.setValue(row, value);
                    }
                }
            }
        }
    }

    /**
     * 获取id值对应的条件对象
     *
     * @param objectMeta
     * @param ids
     * @param forUpdate
     * @param dataSetMeta
     * @return
     */
    private WhereCondition genCondition(IDataObjectMeta objectMeta, Object[] ids, boolean forUpdate,
                                        DataSetMeta dataSetMeta) {
        forUpdate = this.resolveForUpdate(forUpdate);

        StringBuilder sbWhere = new StringBuilder();

        TableMeta rootTableMeta = dataSetMeta.getRootTableMeta();
        FieldMeta pkFieldMeta = rootTableMeta.getPkFieldMeta();
        String pkField = MessageFormat.format("{0}.{1}", rootTableMeta.getName(),
                pkFieldMeta.getDataField().getDbFieldName());
        int sqlType = pkFieldMeta.getDataField().getJDBCTypeNumber();

        if (ids.length == 1) {
            sbWhere.append(pkField);
            sbWhere.append(" = ?");
            return new WhereCondition(sbWhere.toString(), forUpdate, new Object[]{ids[0]}, new int[]{sqlType});
        } else {
            sbWhere.append(pkField);
            sbWhere.append(" IN (");

            int[] argTypes = new int[ids.length];
            for (int i = 0; i < ids.length; i++) {
                sbWhere.append("?,");
                argTypes[i] = sqlType;
            }

            sbWhere.deleteCharAt(sbWhere.length() - 1);
            sbWhere.append(")");

            return new WhereCondition(sbWhere.toString(), forUpdate, ids, argTypes);
        }
    }

    /**
     * 获取查询id的sql语句
     *
     * @param objectMeta
     * @param whereSql
     * @return
     */
    private String genSelectIdSql(IDataObjectMeta objectMeta, String whereSql) {
        IDataField pkField = objectMeta.getPkField();

        StringBuilder sql = new StringBuilder();

        sql.append("SELECT ");
        sql.append(pkField.getDbFieldName());

        sql.append(" FROM ");
        sql.append(objectMeta.getTableName());

        if (!StringUtils.isEmpty(whereSql)) {
            sql.append(" WHERE ").append(whereSql);
        }

        return sql.toString();
    }

    /**
     * 获取指定数据类型的缓存对象
     *
     * @param metaName
     * @return
     */
    private ICache<String, Object> getCache(String metaName) {
        if (this.cacheManager == null) {
            return null;
        }

        ICache<String, Object> cache = this.cacheManager.getCache(metaName);
        if (cache == null) {
            return null;
        }

        DataContractCacheOption cacheOption = DataContractCacheOption.getIfPresent();
        if (cacheOption != null && cacheOption.isCacheDisable(cache.getName())) {
            return null;
        }

        if (TransactionHelper.inTransactionScope()) {
            // 事务内，需要返回一个用于事务隔离的缓存装饰
            return TransactionScopeDataContractCache.create(cache);
        } else {
            return cache;
        }
    }

    /**
     * 访问缓存，尝试获取请求id对应的数据,缓存中如存在，则数据存在tempObjects中，不存在的id则返回
     *
     * @param cache
     * @param objectMeta
     * @param oids
     * @param oidList
     * @param tempObjects
     * @return
     */
    private Set<Object> getUnCacheIds(ICache<String, Object> cache, IDataObjectMeta objectMeta, Object[] oids,
                                      List<Object> oidList, Object[] tempObjects) {
        if (cache == null) {
            return new LinkedHashSet<Object>(oidList);
        }

        Class<?> idClass = objectMeta.getPkField().getFieldClass();
        Set<Object> unCacheIds = new LinkedHashSet<Object>(oids.length, 1);

        // 批量读取缓存
        Set<String> keys = DataContractCacheKeyHelper.genCacheKeys(objectMeta, oids);
        Map<String, ICache.ValueWrapper<Object>> result = cache.getAll(keys);

        // 遍历缓存结果，有数据的项存入tempObjects，不存在的存入unCacheIds
        for (Entry<String, ICache.ValueWrapper<Object>> entry : result.entrySet()) {
            String key = entry.getKey();
            Object id = DataContractCacheKeyHelper.getId(key, idClass);
            Object value = entry.getValue().get();

            if (value != null) {// 当前id已缓存 返回对象
                // 如果对象为NullObject,是为了避免缓存穿透
                if (!NullObject.INSTANCE.equals(value)) {
                    int index = oidList.indexOf(id);
                    if (index > -1) {
                        tempObjects[index] = value;
                    }
                }
            } else {// 标记当前的id在缓存中不存在
                unCacheIds.add(id);
            }
        }

        return unCacheIds;
    }

    private Object[] innerLoadBatchFormDB(IDataObjectMeta objectMeta, DataSetMeta dataSetMeta, Object[] ids,
                                          boolean forUpdate) {
        // 2根据ids构建一个条件对象:WhereCondition,它很简单
        WhereCondition whereCondition = this.genCondition(objectMeta, ids, forUpdate, dataSetMeta);
        // 3.遍历datasetMeta中的所有表，用dataReader进行查询，返回值是一个dataset对象，其封装了N个表的数据
        DataReader dataReader = new DataReader(this.getSqlParser(), this.getSqlExecutor(),
                this.getTransactionTemplate());
        DataSet dataSet = dataReader.readToDataSet(dataSetMeta, whereCondition);

        // 4.根据objectMeta，创建数据对象，将dataset的数据填充到数据对象中
        DataMapper dataMapper = new DataMapper();
        Object[] mapDataObjects = dataMapper.mapDataObjects(dataSet, objectMeta);

        return mapDataObjects;
    }

    private List<RowContract> innerQueryForRows(RowContractMeta rowMeta, String sql, int pageIndex, int pageSize,
                                                Object[] args, int[] argTypes) {
        String sql2exe = sql;

        if (pageIndex > 0 && pageSize > 0) {
            sql2exe = this.getSqlParser().resolvePageSql(sql, pageIndex, pageSize);
        }

        List<RowContract> rows = this.getSqlExecutor().queryForRows(rowMeta, sql2exe, args, argTypes);

        if (rowMeta.isFlatReferenceFields()) {
            this.fillReferenceFields(rowMeta, rows.toArray(new Object[rows.size()]));
            this.flatReferenceFields(rowMeta, rows);
        }

        return rows;
    }

    private Object[] loadBatchFormDB(IDataObjectMeta objectMeta, Object[] ids, boolean forUpdate) {
        ids = this.tryConvertIdValues(ids, objectMeta.getPkField());// 为了后期缓存准备，进行类型转换

        // 1.根据objectMeta创建datasetMeta
        DataSetMeta dataSetMeta = DataSetMeta.create(objectMeta);
        int batchSize = 1000;

        if (ids.length <= batchSize) {
            return this.innerLoadBatchFormDB(objectMeta, dataSetMeta, ids, forUpdate);
        } else {
            List<Object> allObjects = new ArrayList<>(ids.length);
            List<Object[]> idsArr = PageUtils.getAllPageDatas(batchSize, ids);

            for (Object[] partIds : idsArr) {
                Object[] mapDataObjects = this.innerLoadBatchFormDB(objectMeta, dataSetMeta, partIds, forUpdate);
                allObjects.addAll(Arrays.asList(mapDataObjects));
            }

            Class<?> dataContractClass = objectMeta.getDataObjectClass();
            Object[] returnObjects = (Object[]) Array.newInstance(dataContractClass, allObjects.size());

            return allObjects.toArray(returnObjects);
        }
    }

    private Object[] loadBatchFromCache(IDataObjectMeta objectMeta, Object[] ids, boolean forUpdate) {
        ids = this.tryConvertIdValues(ids, objectMeta.getPkField());

        List<Object> idList = Arrays.asList(ids);
        // 存放结果的数组，其索引与ids的索引一致
        Object[] tempObjects = new Object[ids.length];

        // 缓存
        ICache<String, Object> cache = this.getCache(objectMeta.getName());

        // 获取未缓存的id，dataobjects填充已缓存的数据
        Set<Object> unCacheIds = getUnCacheIds(cache, objectMeta, ids, idList, tempObjects);
        if (unCacheIds.size() > 0) {
            // 从数据读取未缓存的数据
            Object[] unCahceObjs = this.loadBatchFormDB(objectMeta, unCacheIds.toArray(), forUpdate);

            Map<String, Object> addToCache = new HashMap<>(ids.length, 1);

            for (Object object : unCahceObjs) {
                Object id = objectMeta.getPkField().getValue(object);
                int index = idList.indexOf(id);
                if (index > -1) {
                    tempObjects[index] = object;
                }

                if (cache != null) {
                    // 准备添加到缓存
                    String key = DataContractCacheKeyHelper.genCacheKey(objectMeta, id);
                    addToCache.put(key, object);

                    // 从未缓存标记集合里删除
                    unCacheIds.remove(id);
                }
            }

            if (cache != null) {
                // 数据库也查不到的数据，标记为NullObject，防止缓存穿透
                for (Object noDataId : unCacheIds) {
                    String key = DataContractCacheKeyHelper.genCacheKey(objectMeta, noDataId);
                    addToCache.put(key, NullObject.INSTANCE);
                }

                // 查到的数据，批量放入缓存
                cache.putAll(addToCache);
            }
        }

        return Stream.of(tempObjects).filter(o -> o != null)
                .toArray(s -> (Object[]) Array.newInstance(objectMeta.getDataObjectClass(), s));
    }

    /**
     * 发布数据写入事件
     *
     * @param writeEvent
     */
    private void publishWriteEvent(DataObjectWriteEvent writeEvent) {
        if (TransactionHelper.inTransactionScope()) {
            TransactionHelper.doAfterCommit(() -> {
                this.applicationEventPublisher.publishEvent(writeEvent);
            });
        } else {
            this.applicationEventPublisher.publishEvent(writeEvent);
        }
    }

    private Object[] queryForObjectsFromCache(IDataObjectMeta objectMeta, String whereSql, boolean forUpdate,
                                              Object[] args, int[] argTypes) {

        String sql = this.genSelectIdSql(objectMeta, whereSql);
        List<Map<String, Object>> list = this.queryForList(sql, args, argTypes);

        IDataField pkField = objectMeta.getPkField();
        String pkName = pkField.getDbFieldName();

        Object[] ids = list.stream().map(map -> map.get(pkName)).toArray();
        Object[] dataobjects = this.loadBatch(objectMeta, ids, true, forUpdate);

        return dataobjects;
    }

    private Object[] queryForObjectsFromDB(IDataObjectMeta objectMeta, String whereSql, boolean forUpdate,
                                           Object[] args, int[] argTypes) {
        forUpdate = this.resolveForUpdate(forUpdate);

        DataSetMeta dataSetMeta = DataSetMeta.create(objectMeta);
        WhereCondition whereCondition = new WhereCondition(whereSql, forUpdate, args, argTypes);

        DataReader dataReader = new DataReader(this.getSqlParser(), this.getSqlExecutor(),
                this.getTransactionTemplate());
        DataSet dataSet = dataReader.readToDataSet(dataSetMeta, whereCondition);

        DataMapper dataMapper = new DataMapper();
        Object[] mapDataObjects = dataMapper.mapDataObjects(dataSet, objectMeta);

        return mapDataObjects;
    }

    private void removeFromCache(IDataObjectMeta objectMeta, Object[] ids) {
        ICache<String, Object> cache = this.getCache(objectMeta.getName());
        if (cache == null) {
            return;
        }

        Set<String> keys = DataContractCacheKeyHelper.genCacheKeys(objectMeta, ids);
        cache.removeAll(keys);
    }

    private boolean resolveForUpdate(boolean forUpdate) {
        return forUpdate && TransactionHelper.inTransactionScope();
    }

    private void setSqlParser(String dataSourceName, IDataSourceManager dataSourceManager) {
        DbInfo dbInfo = dataSourceManager.getDataSourceInfo(dataSourceName);
        if (dbInfo == null) {
            Assert.notNull(dbInfo, "dataSourceName is not found");
        }
        String sqlParserName = StringUtils.uncapitalize(dbInfo.getDbType().getAlias()) + "Parser";
        this.sqlParser = SpringApplicationContext.getInstance().getBean(sqlParserName, ISqlParser.class);
    }

    /**
     * 将外部传入的主键值的类型转换成契约主键规定的类型,并去掉重复值,按原始顺序返回
     *
     * @param oids
     * @param pkMember
     */
    private Object[] tryConvertIdValues(Object[] ids, IDataField pkField) {
        Class<?> fieldClass = pkField.getFieldClass();

        for (int i = 0; i < ids.length; i++) {
            Object id = ids[i];
            if (fieldClass.equals(id.getClass())) {
                continue;
            }

            ids[i] = ConvertUtils.convert(id, fieldClass);
        }

        // 保证原始顺序+去重
        return new LinkedHashSet<>(Arrays.asList(ids)).stream().filter(id -> id != null).toArray();
    }

    /**
     * 迭代保存后的数据
     * <p>
     *
     * <li>对于子表集合，清空之前删除的对象
     * <li>标记所有数据为UnChanged、fromDB为true
     *
     * @author hangwen
     */
    private static class AcceptChangeHandler implements IDataObjectIterateHandler {
        @Override
        public void handleDataCollection(DataCollectionIterateEvent e) {
            // 分析集合属性，清空存放删除id的集合
            ICollectionField field = e.getCollectionField();
            Object parentDataObject = e.getParentObject();

            Collection<IDataContract> removedObjects = field.getRemovedObjects(parentDataObject);
            removedObjects.clear();
        }

        @Override
        public void handleDataObjects(DataObjectIterateEvent e) {
            // 设置对象isFromDB为true,状态为UnChanged
            IDataObjectMeta objectMeta = e.getObjectMeta();
            for (Object dataObj : e.getDataObjects()) {
                if (objectMeta.getState(dataObj).equals(DataObjectState.UnChanged)) {
                    continue;
                }

                objectMeta.setFromDB(dataObj);

                for (IDataField dataField : objectMeta.getDataFields()) {
                    String tableName = dataField.getHostTableName();
                    if (tableName.equals(objectMeta.getTableName())) {
                        continue;
                    }

                    Object val = dataField.getValue(dataObj);
                    if (val == null) {
                        continue;
                    }

                    String subTableName = tableName;
                    objectMeta.setFromDB(dataObj, subTableName);
                }

                objectMeta.setState(dataObj, DataObjectState.UnChanged);
            }
        }

        @Override
        public boolean iterateDataObjectField(IDataObjectField field) {
            return false;
        }
    }

}
